package com.qupeng.demo.kafka.kafkaapache.producer;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MyProducer<K, V> {

    private final static Logger logger = LoggerFactory.getLogger(MyProducer.class);

    public <K, V> KafkaProducer<K, V> createKafkaProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("delivery.timeout.ms", 30000);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        return new KafkaProducer<>(props);
    }

    public void sendAndForget() {
        KafkaProducer<String, String> kafkaProducer = createKafkaProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
        try {
            kafkaProducer.send(record);
        } catch (Exception e) {
            logger.error("", e);
        }
        kafkaProducer.flush();
        kafkaProducer.close();
    }

    public void sendSync() {
        KafkaProducer<String, String> kafkaProducer = createKafkaProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
        try {
            RecordMetadata recordMetadata = kafkaProducer.send(record).get();
        } catch (Exception e) {
            logger.error("", e);
        }
        kafkaProducer.close();
    }

    public void sendAsync() {
        KafkaProducer<String, String> kafkaProducer = createKafkaProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
        kafkaProducer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (null == exception) {

                } else {
                    logger.error("", exception);
                }
                kafkaProducer.close();
            }
        });
    }
}
